package xuecheng.tool.mq.alibaba.rocket;

import org.springframework.util.StringUtils;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;

import lombok.extern.slf4j.Slf4j;
import xuecheng.tool.json.JsonUtils;
import xuecheng.tool.json.JsonUtilsFactory;
import xuecheng.tool.mq.MqListener;

/**
 * rocketmq 消息订阅监听回调自定义实现
 *
 * @author PengXueCheng
 * @version 1.0.0
 */
@Slf4j
class CpxMessageListener<T> implements MessageListener {

    private final JsonUtils JSON_UTILS = JsonUtilsFactory.obtain();

    /**
     * 消息监听回调
     */
    private MqListener<T> listener;
    /**
     * 消息实体类(必须为 Object)
     */
    private Class<T> messageClass;

    CpxMessageListener(MqListener<T> listener, Class<T> messageClass) {
        this.listener = listener;
        this.messageClass = messageClass;
    }

    @Override
    public Action consume(Message message, ConsumeContext context) {
        String topic = message.getTopic();
        String tag = message.getTag();
        String topicAndTag = String.format("%s:%s", topic, tag);
        String msgID = message.getMsgID();
        byte[] body = message.getBody();
        String messageJson = new String(body);
        if (StringUtils.isEmpty(messageJson)) {
            log.warn("接收到消息. 但消息体为空");
            // 消费成功，继续消费下一条消息
            return Action.CommitMessage;
        }
        // 控制台显示的消息
        String consoleShowMsg = messageJson;
        // 控制台消息显示的最大长度
        int maxConsoleShowMsgLen = 100;
        if (messageJson.length() > maxConsoleShowMsgLen) {
            consoleShowMsg = String.format("%s...", messageJson.substring(0, maxConsoleShowMsgLen));
        }
        log.info("接收到消息, topic: {}, msgID: {}, message: {}", topicAndTag, msgID, consoleShowMsg);

        // 消息实体
        T t;
        try {
            t = JSON_UTILS.fromJsonAsObject(messageJson, messageClass);
        } catch (Exception e) {
            log.error(messageJson + "_" + messageClass + "_" + e.getMessage(), e);
            // 消费失败，告知服务器稍后再投递这条消息，继续消费其他消息
            return Action.ReconsumeLater;
        }

        try {
            listener.onMessage(t);
            // 消费成功，继续消费下一条消息
            return Action.CommitMessage;
        } catch (Exception e) {
            log.error(listener.toString() + "_" + t.toString() + "_" + e.getMessage(), e);
            // 消费失败，告知服务器稍后再投递这条消息，继续消费其他消息
            return Action.ReconsumeLater;
        }
    }
}